Maybe we could just put the pair <path, pos> into a fixed array, that seems more memory efficient ?

That would be memory efficient if all of the paths that are equal are also the same reference. I like how using a map doesn't have that assumption, but it probably isn't a big concern.

Why use a delta writer if eq deletes are disabled?

I typically like to use classes that don't need to check configuration in a tight loop. This setting introduces at least one check per row. I'd prefer using either a normal task writer or a delta writer depending on whether deletes are expected in the stream.

Why use a delta writer if eq deletes are disabled?

Because I only want to expose the BaseDeltaWriter to compute engines, I planed to make the BaseRollingWriter & RollingFileWriter & RollingEqDeleteWriter to be private. To implement the compute-engine specific TaskWriter, the only thing we need to do is implementing the asKey and asCopiedKey methods and customizing the policy to dispatch records to DeltaWriter.

How would this set the row? Would we need to keep track of it somehow?

The straightforward way is adding a row field in FilePos which will reference to the inserted old row, but that will hold references of all the inserted rows in a checkpoint. If the row is large while equality fields are small, then the idea way should only keep the equality fields & file-pos in the insertedRowMap, but if we want to attach row when writing pos-delete file then the memory consumption is an issue. I'm considering that maybe we will need an embedded KV lib which could split to disk in future.

Minor: dataWriter should be set to null so that it can be garbage collected and so any further calls to write will fail.

The list equalityFieldIds is only used in this constructor and it is used to create a projection of the schema that is passed in. I think it would be better to pass the delete schema or null in, so that we don't need each writer to create a new projection of the row schema.

This one doesn't need to be public.

I always like to include whatever context is available. Here, it may be helpful to know which partition writer failed. What about using throw new UncheckedIOException("Failed to write position delete file for partition " + partitionKey, e)

I would expect this to call close rather than flush. While close just calls flush so they are equivalent right now, I think using close is better in the long term. If close is modified in the future, it is unlikely that someone will go here and make the same change.

Yeah, make sense ! It's better to use close in this complete method.

Rather than copying the loops, could this call addDataFiles and addDeleteFiles?

The dataFiles() will return an array (I defined it as an array because wanting to avoid the serialization issue for ImmutableList), while the current addDataFiles would only accept Iterable<DataFile>. I think I can add a addDataFiles(DataFile... dataFiles).

I think this should use a StructLikeSet, or else there is no expectation that the returned set will function properly.

Make sense. I will provide a full unit test to cover all those map API.

I think this should be implemented.

It isn't too difficult to implement Map.Entry with a getKey method that calls StructLikeWrapper::get. This method is commonly called on maps. It's even called from putAll above in this class.

Can't this just delegate?

  @Override
  public boolean containsValue(Object value) {
    return wrapperMap.containsValue(value);
  }

This needs to use a new CharSequenceMap or Map<CharSequenceWrapper, List<Long>>.

Why not implement this?

I expected to provide the implemented methods with complete coverage unit tests. But for the purpose providing the PoC solution as soon as possible, I did't have the time to write those tests so left them unsupported now.

Nit: this line doesn't need to change. Can you revert it to avoid commit conflicts?

I think we should separately fix the schema that gets passed to the createWriterFunc. That's a bug.

I don't think this will work because the writer that is returned will be wrapped by a PositionDeleteStructWriter. That would duplicate the position delete struct because this is going to produce a writer for it as well. That's why the schema passed here is a bug, like I noted above.

Yeah, this needs to be fixed. I'm looking through how Avro actually works right now and it is okay because we're calling setSchema a second time from the position writer, which basically discards the original writer with the extra record and rebuilds it with just the row schema.

I'll open a PR to fix it.

Should this also check that there are no delete files?

Yes, it should be emit the deleteFiles to downstream, and commit both delete files and data files into iceberg table by RowDelta API. I think that would be a separate PR to address this.

Nit: this file doesn't need to change.

I think one of the first commits to get this in could make these small changes to both Spark and Flink to update the appender factory. We could do the same for generics as well.

I think this is a bug. It doesn't extract the key fields, it just wraps the row as a StructLike. It should extract the equality fields to produce a key, probably using StructProjection.

I think the tests work because the ID column is the first column in the record.

The tests work because the StructLikeMap has a StructLikeWrapper which will only compare the equality fields, even here we provided the full columns. The name asKey and asCopiedKey are not inappropriate here, asStructLike and asCopiedStructLike will be better.

The tests work because the StructLikeMap has a StructLikeWrapper which will only compare the equality fields

But this happens using the key schema and fields are accessed by position. Wouldn't that fail if the key schema wasn't a prefix of the row schema?

You can use spec.isUnpartitioned().

I'm not familiar with UPDATE_AFTER or UPDATE_BEFORE. Can you help me understand what's going on here?

For CDC events, such as mysql binlog: UPDATE test set a=2, b=2 where a=1, b=1, the flink-cdc-connectors will produce two change logs for it:

-U (a=1, b=1)
+U (a=2,b=2)

The first RowData means we will need to delete the old row (1,1), it's also called a UPDATE_BEFORE row. The second RowData means we will need to insert the new row (2,2), it's also called a UPDATE_AFTER row.

What distinguishes UPDATE_BEFORE from DELETE? Is it that it will be associated with UPDATE_AFTER so the two are combined to form a replace operation?

Is it that it will be associated with UPDATE_AFTER so the two are combined to form a replace operation?

Yes, that's right.

You can supply a checked exception class that will be thrown. That way, you can move the try/catch outside of the run block:

  @Override
  public void close() {
    try {
      Tasks.foreach(deltaWriterMap.values())
          .throwFailureWhenFinished()
          .noRetry()
          .run(RowDataDeltaWriter::close, IOException.class);
    } catch (IOException e) {
      throw new UncheckedIOException("Failed to close writers", e);
    }
  }

And you can also add a callback that receives the exception for each failure with onFailure if you want to log the exceptions individually.

Can you use Assume.assumeTrue? That way it doesn't look like ORC is passing. It shows up as skipped.

Yes, make sense !